package com.atguigu.flink.window.agg;

import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/22
 *
 *     每隔10s，求所有传感器的水位和
 *
 */
public class Demo6_Reduce2
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
           .reduce(new ReduceFunction<WaterSensor>()
                   {
                       @Override
                       public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                           System.out.println("Demo5_Reduce.reduce");
                           value2.setVc(value1.getVc() + value2.getVc());
                           return value2;
                       }
                   },
               //ProcessAllWindowFunction 把ReduceFunction的输出 作为输入
               //ProcessAllWindowFunction  在窗口被触发时，只调用一次
               new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>()
               {
                   @Override
                   public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        //从elements取出最终聚合的结果
                       WaterSensor waterSensor = elements.iterator().next();

                       out.collect(MyUtil.parseTimeWindow(context.window()) +":" + waterSensor.getVc());
                   }
               })
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
